package com.redis.queen;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.TypeReference;
import redis.clients.jedis.Jedis;

import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.UUID;

public class RedisDelayingQueen<T> {
    private static final String BASE_QUEEN_KEY = "DELAYING_QUEEN_KEY:";
    private static final long DEFAULT_GAP = 100L;

    private static final String getLua = "local elements = redis.call('zrangebyscore', KEYS[1], ARGV[1], ARGV[2], 'LIMIT', ARGV[3], ARGV[4])\n" +
            "if  #elements > 0 then\n" +
            "    redis.call('zrem', KEYS[1], elements[1]);\n" +
            "    return elements[1];\n" +
            "else\n" +
            "    return nil;\n" +
            "end";


    private List<String> luaKeys;

    private Jedis jedis;

    private String queenName;

    private String queenKey;

    private Type type = new TypeReference<TaskItem<T>>() {
    }.getType();

    public RedisDelayingQueen(Jedis jedis, String queenName) {
        this.jedis = jedis;
        this.queenName = queenName;
        this.queenKey = BASE_QUEEN_KEY + queenName;
        this.luaKeys = Collections.singletonList(queenKey);
    }

    public void delay(T msg, long seconds) {
        long maturity = System.currentTimeMillis() + seconds * 1000;
        TaskItem<T> taskItem = new TaskItem<>(UUID.randomUUID().toString(), msg);
        jedis.zadd(queenKey, maturity, JSON.toJSONString(taskItem));
    }

    public T get() {
        while (!Thread.interrupted()) {
            List<String> queryArgv = new ArrayList<>();
            queryArgv.add("0");
            queryArgv.add(String.valueOf(System.currentTimeMillis()));
            queryArgv.add("0");
            queryArgv.add("1");
            Object result = jedis.eval(getLua, luaKeys, queryArgv);
            if (result != null) {
                TaskItem<T> item = JSON.parseObject(result.toString(), type);
                return item.getMsg();
            } else {
                try {
                    Thread.sleep(DEFAULT_GAP);
                } catch (InterruptedException ignored) {
                }
            }
        }
        throw new RedisQueenException("fail to get element!");
    }

    public RedisDelayingQueen(String host, int port, String queenName) {
        this(new Jedis(host, port), queenName);
    }

    public void close() {
        if (jedis == null) {
            return;
        }
        jedis.close();
    }
}
